package com.oct.ywsc.job;

import com.alibaba.fastjson2.JSONObject;
import com.oct.ywsc.core.StringUtil;
import com.oct.ywsc.core.bean.RuleConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.ibatis.jdbc.SqlRunner;
import org.apache.ibatis.session.SqlSession;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.util.StringUtils;

import java.sql.SQLException;
import java.util.*;

/**
 * 消息转换规则
 */
@Slf4j
public class TransRule implements ITransRule {

    private final String split_flag = ",";

    private final String param_prefix_flag = "#";

    private RuleConfig ruleConfig;

//    private SqlSession sqlSession;

    private JdbcTemplate jdbcTemplate;


    public TransRule(RuleConfig ruleConfig) {
        this.ruleConfig = ruleConfig;
//        this.sqlSession = sqlSession;
    }

    public String getName(){
        return this.ruleConfig.getTableName()+"#"+this.ruleConfig.getOpType()+"#"+this.ruleConfig.getGzName();
    }

    /**
     * 判断是否匹配该规则
     * @param tableName
     * @param opType
     * @return
     */
    public boolean isMatch(String tableName,String opType){
        return this.ruleConfig.getTableName().equals(tableName)
                && this.ruleConfig.getOptTypeEnum().getOptType().equals(opType);
    }

    public void setJdbcTemplate(JdbcTemplate jdbcTemplate){

        this.jdbcTemplate = jdbcTemplate;
    }

    /**
     * 处理kafka消息并转换为json格式对象
     * @param msg
     * @return
     */
    public List<Map<String,Object>> doTrans(Map<String,String> msg){

        // 拆分sql语句中参数
        List<String> paramList = splitParam();

        // 解析规则sql中 字符串参数


        // 利用kafka中消息替换 规则sql中参数

        // 执行 规则sql
        List<Map<String,Object>> list = execSql(this.ruleConfig.getGzSql(),msg);

//        if (!CollectionUtils.isEmpty(list)){

            return list;
//        }else{
//            log.warn(String.format("规则sql[%s],参数列表[%s]执行结果为空",this.ruleConfig.getGzSql(),JSONObject.toJSONString(msg)));
//            return null;
//        }
    }

    /**
     * 获取发送的topic
     * @return
     */
    public List<String> sendTopic(){
        String topics = this.ruleConfig.getResourceTopic();
        List<String> topicsList = new ArrayList<>();
        if (topics.contains(split_flag)){
            topicsList = Arrays.asList(topics.split(split_flag));
        }else{
            topicsList.add(topics);
        }
        return topicsList;
    }


    /**
     * 获取发送的消息字段
     * @return
     */
    public List<String> sendMsgField(){
        String fields = this.ruleConfig.getResourceFiled();
        if (fields.contains(StringUtil.SPLIT_FLAG)){
            return Arrays.asList(fields.split(split_flag));
        }else{
            return Arrays.asList(fields);
        }
    }


    /**
     * 拆分sql参数字段
     * @return
     */
    private List<String> splitParam(){
        List<String> paramList = null;
        if (StringUtils.hasText(this.ruleConfig.getGzParam() )){
            paramList = new ArrayList<>();
            String[] params = this.ruleConfig.getGzParam().split(split_flag);
            paramList.addAll(Arrays.asList(params));
        }
        return paramList;
    }




    /**
     * 执行sql
     * @param sql
     * @param msg
     * @return
     */
    private List<Map<String,Object>> execSql(String sql, Map<String,String> msg) {

        // 格式化sql
        sql = StringUtil.formatSql(sql,this.splitParam(),msg);

        log.info(String.format("执行规则[%s],sql[%s],参数列表[%s]",this.getName(),sql,JSONObject.toJSONString(msg)));


        List<Map<String,Object>> list = null;
        list = jdbcTemplate.queryForList(sql);

        return list;
    }

}
